package org.example

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object spackstreaming {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  // 5秒统计一次 \1
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //数据处理
    val dStream: ReceiverInputDStream[String] =
      ssc.socketTextStream("192.168.64.129",9999) //   词频统计
    val lines: DStream[String] = dStream.flatMap(_.split(" "))
    val wordDStream: DStream[(String,Int)] = lines.map((_,1))
    val result = wordDStream.reduceByKey(_+_)
    //数据保存
    result.print()
    ssc.start()
    ssc.awaitTermination()
  }
}
